package com.shujia.flink.state

import java.lang

import org.apache.flink.api.common.functions.RuntimeContext
import org.apache.flink.api.common.state.{ListState, ListStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object Demo4ListState {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val studentDS: DataStream[String] = env.socketTextStream("master", 8888)

    /**
      * 实时统计每个班级的平均年龄
      *
      */
    val kvDS: DataStream[(String, Int)] = studentDS.map(stu => {
      val split: Array[String] = stu.split(",")
      val clazz: String = split(4)
      val age: Int = split(2).toInt
      (clazz, age)
    })
    //按照班级分组
    val keyByDS: KeyedStream[(String, Int), String] = kvDS.keyBy(_._1)

    /**
      * process: flink底层的api ,一条一条将数据传到后面的方法中
      * 可以在process操作flinkd的时间，状态，和事件
      *
      */

    val avgAgeDS: DataStream[(String, Double)] = keyByDS.process(new KeyedProcessFunction[String, (String, Int), (String, Double)] {


      /**
        * open： 在processElementz之前执行
        *
        */

      /**
        * 集合状态：为每一个key保存一个集合
        *
        */
      var listState: ListState[Int] = _

      override def open(parameters: Configuration): Unit = {
        //获取flink环境
        val context: RuntimeContext = getRuntimeContext

        //创建集合状态的描述对象
        val listStateDesc = new ListStateDescriptor[Int]("ages", classOf[Int])

        listState = context.getListState(listStateDesc)
      }


      /**
        * processElement: 每一条数据执行一次
        *
        * @param kv  :一条数据
        * @param ctx ：上下文对象
        * @param out ： 用于将数据发送到下游
        */
      override def processElement(kv: (String, Int),
                                  ctx: KeyedProcessFunction[String, (String, Int), (String, Double)]#Context,
                                  out: Collector[(String, Double)]): Unit = {
        val (clazz, age) = kv
        //将每一条数据的年龄保存到集合状态中
        listState.add(age)

        //从状态中获取所有的年龄计算平均值
        val iterable: lang.Iterable[Int] = listState.get()

        //导入一个隐式转换将java的集合转换成scala的集合
        import scala.collection.JavaConversions._
        val ages: List[Int] = iterable.iterator().toList

        val avgAge: Double = ages.sum.toDouble / ages.length

        //将数据发送到下游
        out.collect(clazz, avgAge)
      }
    })

    avgAgeDS.print()


    env.execute()

  }

}
